package app.dwm;

import app.dwd.BaseAppFollowV2;
import bean.*;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import function.DIMAsyncFunction;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;
import util.DimUtil;
import util.GmallSinkUtil;
import util.MyCommonUtils;

import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

import static common.Constant.*;

public class DWMPaymentWideFollowApp extends BaseAppFollowV2 {
    public static void main(String[] args) {
        new DWMPaymentWideFollowApp().init(3333,
                                           1,
                                           "DWMPaymentWideFollowApp",
                                           "DWMPaymentWideFollowApp",
                                           TOPIC_DWM_ORDER_WIDE,
                                           Constant.TOPIC_DWD_PAYMENT_INFO
        );
    }


    @Override
    public void run(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streamKafkaSourcehashMap) {
        DataStreamSource<String> orderWideStream = streamKafkaSourcehashMap.get(TOPIC_DWM_ORDER_WIDE);
        DataStreamSource<String> paymentInfoStream = streamKafkaSourcehashMap.get(TOPIC_DWD_PAYMENT_INFO);

        //intervaljoin前提:keyby,水印
        SingleOutputStreamOperator<PaymentWide> twoStreamIntervalJoin = twoStreamIntervalJoin(orderWideStream, paymentInfoStream);
        GmallSinkUtil.sendToKafka(twoStreamIntervalJoin.map(JSONObject::toJSONString), Constant.TOPIC_DWM_PAYMENT_WIDE);

    }


    private SingleOutputStreamOperator<PaymentWide> twoStreamIntervalJoin(DataStreamSource<String> orderWideStream, DataStreamSource<String> paymentInfoStream) {
        KeyedStream<OrderWide, Long> orderWideLongKeyedStream = orderWideStream
            .map(new MapFunction<String, OrderWide>() {
                @Override
                public OrderWide map(String value) throws Exception {
                    return JSON.parseObject(value, OrderWide.class);
                }
            })
            .assignTimestampsAndWatermarks(WatermarkStrategy
                                               .<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                               .withTimestampAssigner((obj, ts) -> MyCommonUtils.toLongTypeTs(obj.getCreate_time())))
            .keyBy(OrderWide::getOrder_id);

        KeyedStream<PaymentInfo, Long> paymentInfoLongKeyedStream = paymentInfoStream
            .map(new MapFunction<String, PaymentInfo>() {
                @Override
                public PaymentInfo map(String value) throws Exception {
                    return JSON.parseObject(value, PaymentInfo.class);
                }
            })
            .assignTimestampsAndWatermarks(WatermarkStrategy
                                               .<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                               .withTimestampAssigner((obj, ts) -> MyCommonUtils.toLongTypeTs(obj.getCreate_time())))
            .keyBy(PaymentInfo::getOrder_id);

        return paymentInfoLongKeyedStream
            .intervalJoin(orderWideLongKeyedStream)
            .between(Time.minutes(-45), Time.minutes(5))
            .process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                @Override
                public void processElement(PaymentInfo left,
                                           OrderWide right,
                                           Context ctx,
                                           Collector<PaymentWide> out) throws Exception {
                    out.collect(new PaymentWide(left, right));
                }
            });


    }
}
